import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * @Date: 2024/1/8 19:42
 * @Description:
 */
public class FlinkCDC_01_DS {
    public static void main(String[] args) {
        // TODO 1. 初始化流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // TODO 2. 检查点
        // 2.1 启用检查点
        env.enableCheckpointing(10 * 1000L);
        // 2.2 设置相邻两次检查点最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30 * 1000L);
        // 2.3 设置取消Job检查点的清理模式
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
        );

        // 2.4 设置状态后端类型
        env.setStateBackend(new HashMapStateBackend());
        // 2.5 设置检查点存储路径
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop100:8020/flinkCDC");
        // 2.6 设置HDFS用户名
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        // TODO 3. 创建Flink-MYSQL-CDC的Source
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop100")
                .port(3306)
                .databaseList("financial_lease_config") // set captured database
                .tableList("financial_lease_config.table_process") // set captured table
                .username("root")
                .password("000000")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .startupOptions(StartupOptions.initial())
                .build();

        // TODO 4. 使用CDC Source从MySQL读取数据
        DataStreamSource<String> mysqlDS =
                env.fromSource(
                        mySqlSource,
                        WatermarkStrategy.noWatermarks(),
                        "MysqlSource");

        // TODO 5. 打印输出
        mysqlDS.print();
        // TODO 6. 执行任务
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
